package com.lifq.flink.source

import com.fasterxml.jackson.databind.ObjectMapper
import org.apache.flink.streaming.api.functions.source.SourceFunction
import org.apache.log4j.Logger

import java.time.LocalDateTime
import java.time.format.DateTimeFormatter
import java.util.Locale
import scala.util.Random

/**
 * @Author: LiFuqiang
 * @Date: 2023/4/10 11:06
 * @Version 1.0 
 * @description
 */
class MockDataGenerator extends SourceFunction[String] {
  
  var isRunning = true
  var data = "0,0,0,1,1,1,1,2,2,2,5,5,5,5,6,5,5,5,6,6,3,4,2,1,5,5,0,2,2,0,0,0,1,1,1,1,2,2,2,5,5,5,5,6,5,5,5,6,6,3,4,2,1,5,5,0,2,2".split(",")
  var names = "Jim,James,Susan,Tom,Joe,Trump,Biden,Bush".split(",")
  var id = 0


  override def run(sourceContext: SourceFunction.SourceContext[String]): Unit = {
    val formatter = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss", Locale.CHINA)
    while (isRunning) {
      id = id + 1
      val index = Random.nextInt(8)
      val name = names(index)
      val date = LocalDateTime.now().format(formatter)
      val msg = id + "," + name + "," + date
      sourceContext.collect(msg)
      Thread.sleep(200)
    }
  }

  override def cancel(): Unit = {
    isRunning = false
  }
}

object MockDataGenerator{
  def main(args: Array[String]): Unit = {
    val date = LocalDateTime.now()
    val formatter = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss", Locale.CHINA)
    val dtStr = formatter.format(date)
    println(dtStr)
  }
}
